//! 基于Crossbeam的高性能事件分发器
//! 
//! 使用成熟的crossbeam库提供线程安全的事件分发功能

use crate::event::{DynamicEvent, DynamicEventHandler, EventResult};

use crossbeam::channel::{unbounded, Receiver, Sender};
use std::sync::{Arc, RwLock};
use fxhash::FxHashMap;
use std::thread;
use std::time::Duration;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

/// Crossbeam分发器配置
#[derive(Debug, Clone)]
pub struct CrossbeamDispatcherConfig {
    /// Worker线程数量
    pub worker_count: usize,
    /// 是否启用统计信息收集
    pub enable_stats: bool,
    /// Worker线程名称前缀
    pub worker_name_prefix: String,
}

impl Default for CrossbeamDispatcherConfig {
    fn default() -> Self {
        Self {
            worker_count: num_cpus::get(),
            enable_stats: true,
            worker_name_prefix: "crossbeam-worker".to_string(),
        }
    }
}

/// 分发器统计信息
#[derive(Debug, Clone)]
pub struct CrossbeamDispatcherStats {
    pub events_sent: u64,
    pub events_processed: u64,
    pub events_failed: u64,
    pub workers_active: usize,
    pub queue_length: usize,
}

/// 基于Crossbeam的事件分发器
pub struct CrossbeamEventDispatcher {
    config: CrossbeamDispatcherConfig,
    event_sender: Sender<DynamicEvent>,
    event_receiver: Receiver<DynamicEvent>,
    handlers: Arc<RwLock<FxHashMap<String, Vec<Arc<dyn DynamicEventHandler>>>>>,
    running: Arc<AtomicBool>,
    worker_handles: Vec<thread::JoinHandle<()>>,
    
    // 统计信息
    stats_events_sent: Arc<AtomicU64>,
    stats_events_processed: Arc<AtomicU64>,
    stats_events_failed: Arc<AtomicU64>,
}

/// Crossbeam事件发送器（线程安全）
#[derive(Clone)]
pub struct CrossbeamEventSender {
    sender: Sender<DynamicEvent>,
    stats_events_sent: Arc<AtomicU64>,
}

impl CrossbeamEventSender {
    /// 发布事件
    pub fn publish(&self, event: DynamicEvent) -> Result<(), DynamicEvent> {
        match self.sender.send(event) {
            Ok(()) => {
                self.stats_events_sent.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            Err(crossbeam::channel::SendError(event)) => Err(event),
        }
    }
    
    /// 尝试发布事件（非阻塞）
    pub fn try_publish(&self, event: DynamicEvent) -> Result<(), DynamicEvent> {
        match self.sender.try_send(event) {
            Ok(()) => {
                self.stats_events_sent.fetch_add(1, Ordering::Relaxed);
                Ok(())
            }
            Err(crossbeam::channel::TrySendError::Full(event)) => Err(event),
            Err(crossbeam::channel::TrySendError::Disconnected(event)) => Err(event),
        }
    }
}

impl CrossbeamEventDispatcher {
    /// 创建新的分发器
    pub fn new(config: CrossbeamDispatcherConfig) -> Self {
        let (event_sender, event_receiver) = unbounded();
        
        Self {
            config,
            event_sender,
            event_receiver,
            handlers: Arc::new(RwLock::new(FxHashMap::default())),
            running: Arc::new(AtomicBool::new(false)),
            worker_handles: Vec::new(),
            stats_events_sent: Arc::new(AtomicU64::new(0)),
            stats_events_processed: Arc::new(AtomicU64::new(0)),
            stats_events_failed: Arc::new(AtomicU64::new(0)),
        }
    }
    
    /// 注册事件处理器
    pub fn register_handler(&mut self, event_type: &str, handler: Arc<dyn DynamicEventHandler>) {
        let mut handlers = self.handlers.write().unwrap();
        handlers.entry(event_type.to_string())
            .or_default()
            .push(handler);
    }
    
    /// 获取事件发送器
    pub fn get_sender(&self) -> CrossbeamEventSender {
        CrossbeamEventSender {
            sender: self.event_sender.clone(),
            stats_events_sent: self.stats_events_sent.clone(),
        }
    }
    
    /// 启动分发器
    pub fn start(&mut self) {
        if self.running.load(Ordering::Relaxed) {
            tracing::warn!("CrossbeamEventDispatcher已经在运行中");
            return;
        }
        
        self.running.store(true, Ordering::Relaxed);
        tracing::info!("启动CrossbeamEventDispatcher，工作线程数: {}", self.config.worker_count);
        
        // 启动worker线程
        for worker_id in 0..self.config.worker_count {
            let running = self.running.clone();
            let receiver = self.event_receiver.clone();
            let handlers = self.handlers.clone();
            let stats_processed = self.stats_events_processed.clone();
            let stats_failed = self.stats_events_failed.clone();
            let worker_name = format!("{}-{}", self.config.worker_name_prefix, worker_id);
            
            let handle = thread::Builder::new()
                .name(worker_name.clone())
                .spawn(move || {
                    tracing::debug!("Worker线程 {} 启动", worker_name);
                    
                    while running.load(Ordering::Relaxed) {
                        match receiver.recv_timeout(Duration::from_millis(100)) {
                            Ok(event) => {
                                let event_type = event.event_type().to_string();
                                let handlers_guard = handlers.read().unwrap();
                                
                                if let Some(handler_list) = handlers_guard.get(&event_type) {
                                    if let Some(first_handler) = handler_list.first() {
                                        match first_handler.handle_dynamic(&event) {
                                            EventResult::Success => {
                                                stats_processed.fetch_add(1, Ordering::Relaxed);
                                            }
                                            EventResult::Failed(_) => {
                                                stats_failed.fetch_add(1, Ordering::Relaxed);
                                            }
                                            _ => {}
                                        }
                                        // 只处理第一个匹配的handler
                                    }
                                } else {
                                    tracing::debug!("没有找到事件类型 '{}' 的处理器", event_type);
                                }
                            }
                            Err(crossbeam::channel::RecvTimeoutError::Timeout) => {
                                // 超时，继续检查running状态
                                continue;
                            }
                            Err(crossbeam::channel::RecvTimeoutError::Disconnected) => {
                                tracing::debug!("Worker线程 {} 接收到断开信号", worker_name);
                                break;
                            }
                        }
                    }
                    
                    tracing::debug!("Worker线程 {} 退出", worker_name);
                })
                .expect("Failed to spawn worker thread");
            
            self.worker_handles.push(handle);
        }
        
        tracing::info!("CrossbeamEventDispatcher启动完成");
    }
    
    /// 停止分发器
    pub fn stop(&mut self) {
        if !self.running.load(Ordering::Relaxed) {
            return;
        }
        
        tracing::info!("停止CrossbeamEventDispatcher");
        
        // 设置停止标志
        self.running.store(false, Ordering::Relaxed);
        
        // 等待所有worker线程结束
        while let Some(handle) = self.worker_handles.pop() {
            match handle.join() {
                Ok(()) => tracing::debug!("Worker线程正常退出"),
                Err(e) => tracing::error!("Worker线程异常退出: {:?}", e),
            }
        }
        
        tracing::info!("CrossbeamEventDispatcher已停止");
    }
    
    /// 获取统计信息
    pub fn get_stats(&self) -> CrossbeamDispatcherStats {
        CrossbeamDispatcherStats {
            events_sent: self.stats_events_sent.load(Ordering::Relaxed),
            events_processed: self.stats_events_processed.load(Ordering::Relaxed),
            events_failed: self.stats_events_failed.load(Ordering::Relaxed),
            workers_active: self.worker_handles.len(),
            queue_length: self.event_receiver.len(),
        }
    }
    
    /// 检查是否正在运行
    pub fn is_running(&self) -> bool {
        self.running.load(Ordering::Relaxed)
    }
    
    /// 获取队列长度
    pub fn queue_length(&self) -> usize {
        self.event_receiver.len()
    }
}

// 实现Drop trait确保资源正确清理
impl Drop for CrossbeamEventDispatcher {
    fn drop(&mut self) {
        if self.running.load(Ordering::Relaxed) {
            tracing::warn!("CrossbeamEventDispatcher在Drop时仍在运行，自动停止");
            self.stop();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{Event, EventPayload};
    use std::sync::atomic::AtomicU32;
    use std::time::Instant;
    
    /// 测试负载
    #[derive(Debug, Clone)]
    struct TestPayload {
        message: String,
        id: u64,
    }
    
    impl TestPayload {
        fn new(message: String, id: u64) -> Self {
            Self { message, id }
        }
    }
    
    impl EventPayload for TestPayload {
        fn payload_type(&self) -> &'static str {
            "crossbeam_test"
        }
        
        fn serialize(&self) -> Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>> {
            Ok(format!("{}:{}", self.message, self.id).into_bytes())
        }
        
        fn size(&self) -> usize {
            self.message.len() + 8
        }
    }
    
    /// 测试处理器
    struct TestHandler {
        name: String,
        processed_count: Arc<AtomicU32>,
    }
    
    impl TestHandler {
        fn new(name: String) -> Self {
            Self {
                name,
                processed_count: Arc::new(AtomicU32::new(0)),
            }
        }
        
        fn get_processed_count(&self) -> u32 {
            self.processed_count.load(Ordering::Relaxed)
        }
    }
    
    impl DynamicEventHandler for TestHandler {
        fn handle_dynamic(&self, _event: &DynamicEvent) -> EventResult {
            self.processed_count.fetch_add(1, Ordering::Relaxed);
            EventResult::Success
        }
        
        fn name(&self) -> &str {
            &self.name
        }
        
        fn can_handle(&self, event_type: &str) -> bool {
            event_type == "crossbeam_test"
        }
    }
    
    #[test]
    fn test_crossbeam_dispatcher_basic() {
        tracing_subscriber::fmt()
            .with_max_level(tracing::Level::DEBUG)
            .try_init()
            .ok();
        
        let config = CrossbeamDispatcherConfig {
            worker_count: 2,
            enable_stats: true,
            worker_name_prefix: "test-worker".to_string(),
        };
        
        let mut dispatcher = CrossbeamEventDispatcher::new(config);
        let handler = Arc::new(TestHandler::new("test_handler".to_string()));
        
        dispatcher.register_handler("crossbeam_test", handler.clone());
        dispatcher.start();
        
        // 发送测试事件
        let sender = dispatcher.get_sender();
        const MESSAGE_COUNT: u64 = 100;
        
        for i in 0..MESSAGE_COUNT {
            let payload = TestPayload::new(format!("message_{i}"), i);
            let event = Event::new(
                "crossbeam_test",
                Box::new(payload) as Box<dyn EventPayload>
            );
            
            sender.publish(event).expect("Failed to send event");
        }
        
        // 等待处理完成
        let start = Instant::now();
        while handler.get_processed_count() < MESSAGE_COUNT as u32 {
            if start.elapsed() > Duration::from_secs(5) {
                panic!("Timeout waiting for events to be processed");
            }
            thread::sleep(Duration::from_millis(10));
        }
        
        let stats = dispatcher.get_stats();
        println!("统计信息: {stats:?}");
        
        assert_eq!(stats.events_sent, MESSAGE_COUNT);
        assert_eq!(stats.events_processed, MESSAGE_COUNT);
        assert_eq!(stats.events_failed, 0);
        assert_eq!(handler.get_processed_count(), MESSAGE_COUNT as u32);
        
        dispatcher.stop();
        assert!(!dispatcher.is_running());
    }
    
    #[test]
    fn test_crossbeam_dispatcher_concurrent() {
        let config = CrossbeamDispatcherConfig {
            worker_count: 4,
            enable_stats: true,
            worker_name_prefix: "concurrent-worker".to_string(),
        };
        
        let mut dispatcher = CrossbeamEventDispatcher::new(config);
        let handler = Arc::new(TestHandler::new("concurrent_handler".to_string()));
        
        dispatcher.register_handler("crossbeam_test", handler.clone());
        dispatcher.start();
        
        let sender = dispatcher.get_sender();
        const PRODUCER_COUNT: usize = 4;
        const MESSAGES_PER_PRODUCER: u64 = 250;
        const TOTAL_MESSAGES: u64 = PRODUCER_COUNT as u64 * MESSAGES_PER_PRODUCER;
        
        // 启动多个生产者线程
        let mut producer_handles = vec![];
        
        for producer_id in 0..PRODUCER_COUNT {
            let sender_clone = sender.clone();
            let handle = thread::spawn(move || {
                for i in 0..MESSAGES_PER_PRODUCER {
                    let msg_id = producer_id as u64 * MESSAGES_PER_PRODUCER + i;
                    let payload = TestPayload::new(
                        format!("producer_{producer_id}_message_{i}"), 
                        msg_id
                    );
                    let event = Event::new(
                        "crossbeam_test",
                        Box::new(payload) as Box<dyn EventPayload>
                    );
                    
                    sender_clone.publish(event).expect("Failed to send event");
                }
            });
            producer_handles.push(handle);
        }
        
        // 等待所有生产者完成
        for handle in producer_handles {
            handle.join().unwrap();
        }
        
        // 等待所有消息处理完成
        let start = Instant::now();
        while handler.get_processed_count() < TOTAL_MESSAGES as u32 {
            if start.elapsed() > Duration::from_secs(10) {
                panic!("Timeout waiting for concurrent events to be processed");
            }
            thread::sleep(Duration::from_millis(10));
        }
        
        let stats = dispatcher.get_stats();
        println!("并发统计信息: {stats:?}");
        
        assert_eq!(stats.events_sent, TOTAL_MESSAGES);
        assert_eq!(stats.events_processed, TOTAL_MESSAGES);
        assert_eq!(handler.get_processed_count(), TOTAL_MESSAGES as u32);
        
        dispatcher.stop();
    }
}